package com.example.watermark;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

/**
 * Created with IntelliJ IDEA.
 * ClassName: MyPeriodWatermarkGenerator
 * Package: com.example.watermark
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-20
 * Time: 14:01
 */

public class MyPeriodWatermarkGenerator<T> implements WatermarkGenerator<T> {

    private long delayTs; //乱序程度
    private long maxTs;  //存在最大值  事件时间

    public MyPeriodWatermarkGenerator(long delayTs) {
        this.delayTs = delayTs;
        this.maxTs = Long.MAX_VALUE + this.delayTs + 1;
    }

    /**
     * 每条数据来 都会调用这个方法 用来获取最大事件时间 保存下来
     *
     * @param event
     * @param eventTimestamp 提取到的事件时间
     * @param output
     */
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        //取最大值的方法 even和maxTs比较取最大值
        maxTs = Math.max(maxTs, eventTimestamp);
        System.out.println("调用了onEvent方法 获取目前为止的最大时间戳 MAX= " + maxTs);
    }

    /**
     * 周期调用这个方法 反射watermark
     * @param output
     */
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(maxTs - delayTs - 1));
        System.out.println("调用周期性调用的方法onPeriodicEmit watermark=" + (maxTs-delayTs-1));
    }
}
